package com.atguigu.flink.sqlfunction;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.*;

/**
 * Created by Smexy on 2023/2/6
 *
 *      UDTF: 输入1行，输出N行N列
 *
 *      举例：输入1行，将数据按照_炸裂，输出
 *              输入sensor_1  输出 sensor
 *                                  1
 */
public class Demo3_TableFunction
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        String createTableSql = "create table t1( id string,ts bigint ,vc int ) with(" +
            "                    'connector' = 'filesystem' ," +
            "                    'path' = 'data/func/sensor.txt' , " +
            "                    'format' = 'csv'  " +
            "                    )";

        //执行建表(写操作)
        tableEnvironment.executeSql(createTableSql);

        //把t1变为Table对象
        Table table = tableEnvironment.from("t1");

        //1.先创建一个函数对象
        MyUDTF myFunc = new MyUDTF();

        //注册函数，再用
        tableEnvironment.createTemporaryFunction("a",myFunc);

        /*
                必须使用join的语法来使用 UDTF。

                在hive里面如何使用UDTF?
                        select  from  t1 lateral view UDTF函数()  tmp as col1,col2
                        lateral view 本质是 join!
         */

        //使用方式1，在TableAPI中，用函数名调用
       /* table
            // 内连接
            //.joinLateral(call("a",$("id")))
            //左外连接  取数据表的全部作为左表
            .leftOuterJoinLateral(call("a",$("id")))
            .select($("id"),$("word"))
            .execute().print();*/

        //使用方式2，在SQL中，用函数名调用
        //内连接
        //tableEnvironment.sqlQuery("select id,word from t1, LATERAL TABLE(a(id) ) ").execute().print();

        //左外连接
        //tableEnvironment.sqlQuery("select id,word from t1 LEFT JOIN  LATERAL TABLE(a(id) ) ON TRUE ").execute().print();

        //内连接写法2
        tableEnvironment.sqlQuery("select id,word from t1 INNER JOIN  LATERAL TABLE(a(id) ) ON TRUE ").execute().print();


    }

    /*
       TableFunction<T> 是输出的一行的类型
     */
    //声明输出每行的所有列元数据
    @FunctionHint(output = @DataTypeHint("ROW<word STRING>"))
    public static class MyUDTF extends TableFunction<Row>
    {
        /*
            有方法是函数的处理逻辑
                方法名字必须是 eval
                不同于标量函数，表值函数的求值方法本身不包含返回类型，而是通过 collect(T) 方法来发送要输出的行

             @DataTypeHint:加在方法的参数上，声明入参类型，提高编译器编译的速度

             如果输出行只包含 1 列，会省略结构化信息并生成标量值。如果输出的每一行有多个列，需要声明每一列的元数据(列名和类型)
         */
        public void eval(@DataTypeHint("STRING") String o) {
           //如果是sensor_3不做处理
            if (!o.equals("sensor_3")){

                String[] words = o.split("_");
                //输出第一行
                collect(Row.of(words[0]));
                //输出第二行
                collect(Row.of(words[1]));
            }
        }

    }
}
